#include "EventLoop.h"
#include <sys/eventfd.h>
#include <stdlib.h>
#include <unistd.h>
#include <assert.h>
#include <memory>

#include "Logger.h"
#include "Poller.h"
#include "Channel.h"

// 防止一个线程创建多个EventLoop
__thread EventLoop* t_loopInthisThread = nullptr;

// 设置默认的Poller IO复用接口的超时时间
const int KPollTimeMs = 1000;


// 用来唤醒subReactor处理新的channel
int createEventfd()
{
    // EF_NONBLOCK 非阻塞   EFD_CLOEXEC 子进程在执行exec后会自动关闭文件描述符
    int evtfd = eventfd(0,EFD_NONBLOCK | EFD_CLOEXEC);
    if(evtfd < 0)
    {
        LOG_FATAL("Failed in eventfd\n");
        abort();
    }
    return evtfd;
}

EventLoop::EventLoop()
    : looping_(false),
      quit_(false),
      callingPendingFunctors_(false),
      threadId_(CurrentThread::tid()),
      poller_(Poller::newDefaultPoller(this)),
      weakupFd_(createEventfd()),
      weakupChannel_(new Channel(this, weakupFd_)),
      currentActiveChannel_(nullptr)
{
    LOG_DEBUG("EventLoop create %p in thread %d \n",this,threadId_);
    if(t_loopInthisThread) 
    {
        LOG_FATAL("Another EventLoop %p exists in this thread %d \n",t_loopInthisThread,threadId_);
    }
    else
    {
        t_loopInthisThread = this;
    }

    // 设置weakupFd的事件类型及发生事件后的回调操作
    weakupChannel_->setReadCallback(std::bind(&EventLoop::handleRead,this));    
    // 每一个EventLoop都会监听weakChannel的EPOLLIN读事件
    weakupChannel_->enableReading();
}

EventLoop::~EventLoop()
{
    weakupChannel_->disableAll();
    weakupChannel_->remove();
    close(weakupFd_);
    t_loopInthisThread = nullptr;
}

void EventLoop::loop()
{
    assert(!looping_);
    looping_ = true;
    quit_ = false;
    LOG_DEBUG("EventLoop %p start looping",this);
    while(!quit_)
    {
        activeChannels_.clear();
        pollReturnTime_ = poller_->poll(KPollTimeMs,&activeChannels_);
        for(auto channel : activeChannels_)
        {
            currentActiveChannel_ = channel;
            currentActiveChannel_->handleEvent(pollReturnTime_);
        }
        currentActiveChannel_ = nullptr;
        doPendingFunctors();
    }
    LOG_DEBUG("EventLoop %p stop looping",this);
    looping_ = false;
}

void EventLoop::quit()
{
    quit_ = true;
    // 如果在其他线程中调用了quit，会唤醒主线程，调用quit
    if(!isInLoopThread())
    {
        weakup();
    }
}

void EventLoop::runInLoop(Functor cb)
{
    if(isInLoopThread())
    {
        cb();
    }
    else
    {
        // 如果不在当前的线程，需要唤醒线程所在的loop，执行cb
        queueInLoop(std::move(cb));
    }
}

void EventLoop::queueInLoop(Functor cb)
{
    {
        std::unique_lock<std::mutex> lock(mutex_);
        pendingFunctors_.emplace_back(cb);
    }

    // 如果不是当前的线程 或者  需要唤醒loop所在的线程
    // callingPeingFunctors_ 表示当前loop已经在执行回调了，但是loop又有了新的回调,所以需要继续唤醒执行新的回调函数
    if(!isInLoopThread() || callingPendingFunctors_)
    {
        weakup();
    }
}

// 用来唤醒loop所在的线程， 向weakFd写数据
void EventLoop::weakup()
{
    uint64_t one = 1;
    ssize_t n = write(weakupFd_, &one, sizeof one);
    if (n != sizeof one)
    {
        LOG_ERROR("EventLoop::weakup() write %d bytes instead of 8", n);
    }
}

void EventLoop::updateChannel(Channel *channel)
{
    poller_->updateChannel(channel);
}

void EventLoop::removeChannel(Channel *channel)
{
    poller_->removeChannel(channel);
}

bool EventLoop::hasChannel(Channel *channel)
{
    return poller_->hasChannel(channel);
}

void EventLoop::handleRead()
{
    uint64_t one = 1;
    ssize_t n = read(weakupFd_, &one, sizeof one);
    if (n != sizeof one)
    {
        LOG_ERROR("EventLoop::handleRead() reads %d bytes instead of 8", n);
    }
}

void EventLoop::doPendingFunctors()
{
    std::vector<Functor> functors;
    callingPendingFunctors_ = true;

    {
        std::unique_lock<std::mutex> lock(mutex_);
        functors.swap(pendingFunctors_);
    }

    for(auto functor : functors)
    {
        functor();
    }

    callingPendingFunctors_ = false;
}
